package com.settlement.system.service.impl;

import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.settlement.system.common.constant.KafkaConstants;
import com.settlement.system.common.constant.SystemConstants;
import com.settlement.system.common.util.DateUtils;
import com.settlement.system.common.util.NumberUtils;
import com.settlement.system.common.util.StringUtils;
import com.settlement.system.handler.disruptor.MessageVo;
import com.settlement.system.mapper.WmTradeDetailMapper;
import com.settlement.system.model.entity.SysImportErrorDetail;
import com.settlement.system.model.entity.SysStore;
import com.settlement.system.model.entity.WmNoRecord;
import com.settlement.system.model.entity.WmTradeDetail;
import com.settlement.system.model.form.WmTradeDetailForm;
import com.settlement.system.model.query.WmTradeDetailQuery;
import com.settlement.system.model.vo.WmTradeDetailImportVo;
import com.settlement.system.service.*;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * <p>
 * 微盟交易明细 服务实现类
 * </p>
 *
 * @author huangrensen
 * @since 2024-06-18
 */
@Service
@RequiredArgsConstructor
public class WmTradeDetailServiceImpl extends ServiceImpl<WmTradeDetailMapper, WmTradeDetail> implements WmTradeDetailService {

    private final Logger logger = LoggerFactory.getLogger(WmTradeDetailServiceImpl.class);
    private final SysStoreService sysStoreService;

    private final RedisTemplate redisTemplate;

    private final SysImportErrorDetailService importErrorDetailService;

    private final WmNoRecordService wmNoRecordService;

    private final SnowflakeIdWorker snowflakeIdWorker = SnowflakeIdWorker.getInstance();

    @Override
    public Page<WmTradeDetail> getPage(WmTradeDetailQuery queryParams) {
        // 查询参数
        int pageNum = queryParams.getPageNum();
        int pageSize = queryParams.getPageSize();
        // 拼接月份列表
        List<String> monthList = new ArrayList<>();
        String endMonthStr = queryParams.getEndMonth().replaceAll("-", "");
        LocalDate endMonth = LocalDate.parse(endMonthStr + "01", DateTimeFormatter.ofPattern("yyyyMMdd"));
        LocalDate startTime = LocalDate.parse(queryParams.getStartMonth().replaceAll("-", "") + "01", DateTimeFormatter.ofPattern(
                "yyyyMMdd"));
        // 获取已经生成的表名
        Object object = redisTemplate.opsForValue().get(SystemConstants.WM_TRADE_DETAIL_TABLE);
        List<String> tableNameList;
        if (object != null) {
            tableNameList = (List<String>) object;
        } else {
            tableNameList = this.baseMapper.getAllTable();
            // 添加缓存
            redisTemplate.opsForValue().set(SystemConstants.WM_TRADE_DETAIL_TABLE, tableNameList, 1, TimeUnit.HOURS);
        }
        // 循环追加月份
        while (!startTime.isAfter(endMonth)) {
            monthList.add(startTime.format(DateTimeFormatter.ofPattern("yyyyMM")));
            startTime = startTime.plusMonths(1);
        }
        // 根据月份判断数据库表是否存在
        List<String> existTableList = new ArrayList<>();
        String tableName;
        for (String month : monthList) {
            tableName = "wm_trade_detail_%s".formatted(month);
            if (tableNameList.contains(tableName)) {
                existTableList.add(tableName);
            }
        }
        // 查询数据
        Page<WmTradeDetail> page = new Page<>(pageNum, pageSize);
        if (existTableList.isEmpty()) {
            return page;
        }
        queryParams.setTableNameList(existTableList);
        Page<WmTradeDetail> mapperPage = this.baseMapper.getPage(page, queryParams);
        SysStore sysStore;
        for (WmTradeDetail record : mapperPage.getRecords()) {
            sysStore = sysStoreService.getById(record.getStoreId());
            if (Objects.nonNull(sysStore)) {
                record.setStoreName(sysStore.getCommercePlatformStoreName());
            }
        }
        return mapperPage;
    }

    @Override
    public boolean saveWmTradeDetail(WmTradeDetailForm wmTradeDetailForm) {
        WmTradeDetail wmTradeDetail = new WmTradeDetail();
        BeanUtils.copyProperties(wmTradeDetailForm, wmTradeDetail);
        wmTradeDetail.setCreateTime(null);
        return this.saveOrUpdate(wmTradeDetail);
    }

    @Override
    public WmTradeDetailForm getWmTradeDetailFormData(Long id) {
        WmTradeDetail wmTradeDetail = this.getById(id);
        WmTradeDetailForm wmTradeDetailForm = new WmTradeDetailForm();
        BeanUtils.copyProperties(wmTradeDetail, wmTradeDetailForm);
        return wmTradeDetailForm;
    }

    @Override
    public boolean deleteWmTradeDetail(String ids) {
        Assert.isTrue(StringUtils.hasText(ids), "删除的数据为空");
        List<Long> idArray = Arrays.stream(ids.split(",")).map(Long::parseLong).collect(Collectors.toList());
        return this.removeByIds(idArray);
    }


    @KafkaListener(id = KafkaConstants.HWG_TOPIC_WM_TRADE_DETAIL, topics = KafkaConstants.HWG_TOPIC_WM_TRADE_DETAIL, groupId = "yzConsumerGroup")
    public void importData(List<ConsumerRecord<String, String>> records, Acknowledgment acknowledgment) {
        logger.info("【{}】当前消费者本次拉取的数据量为：{}", KafkaConstants.HWG_TOPIC_WM_TRADE_DETAIL, (Objects.isNull(records) ? 0 : records.size()));
        if (Objects.isNull(records) || records.isEmpty()) {
            logger.error("【{}】无消费数据...", KafkaConstants.HWG_TOPIC_WM_TRADE_DETAIL);
        }
        String orderNo = "";
        for (ConsumerRecord<String, String> record : records) {
            String importVoStr = record.value();
            try {
                WmTradeDetailImportVo importVo = JSONObject.parseObject(importVoStr, WmTradeDetailImportVo.class);
                orderNo = importVo.getTradeNo();
                this.importData(Collections.singletonList(importVo));
            } catch (Exception e) {
                logger.info("消费失败：{}", e.getMessage());
                // 保存失败数据
                SysImportErrorDetail detail = new SysImportErrorDetail();
                detail.setPlatformAndType("微盟交易明细");
                detail.setDataText(importVoStr);
                detail.setOrderNo(orderNo);
                detail.setQueueName(KafkaConstants.HWG_TOPIC_WM_TRADE_DETAIL);
                detail.setRemark(e.getMessage());
                importErrorDetailService.checkAndCreate(detail);
            }
        }
        // 确认消费
        acknowledgment.acknowledge();
    }

    @Override
    public void importData(List<WmTradeDetailImportVo> voList) {
        redisTemplate.delete(SystemConstants.WM_TRADE_DETAIL_TABLE);
        List<String> billMonthList = voList.stream().map(WmTradeDetailImportVo::getBillMonth).distinct().toList();
        // 创建表
        for (String billMonth : billMonthList) {
            if (!StringUtils.hasText(this.baseMapper.existTable("wm_trade_detail_%s".formatted(billMonth)))) {
                this.baseMapper.createTable("wm_trade_detail_%s".formatted(billMonth));
            }
        }
        List<WmTradeDetail> wmTradeDetailList = new ArrayList<>();
        for (WmTradeDetailImportVo importVo : voList) {
            WmTradeDetail wmTradeDetail = new WmTradeDetail();
            BeanUtils.copyProperties(importVo, wmTradeDetail);
            wmTradeDetail.setId(snowflakeIdWorker.nextId(importVo.getBillMonth()));
            // 交易时间
            wmTradeDetail.setTradeTime(DateUtils.convertStringToLocalDateTime(importVo.getTradeTime()));
            // 交易金额
            wmTradeDetail.setTradeAmount(NumberUtils.convertBigDecimal(importVo.getTradeAmount()));
            wmTradeDetail.setTableName("wm_trade_detail_%s".formatted(importVo.getBillMonth()));
            wmTradeDetail.setStoreId(importVo.getStoreId());
            wmTradeDetailList.add(wmTradeDetail);
        }
        MessageVo vo;
        List<WmTradeDetail> itemList = new ArrayList<>();
        List<WmNoRecord> wmNoRecordList = new ArrayList<>();
        // 创建
        WmNoRecord wmNoRecord;
        for (WmTradeDetail item : wmTradeDetailList) {
            item.cutString();
            this.baseMapper.insertData(item);
            itemList.add(item);
            // 微盟交易单号记录表
            wmNoRecord = new WmNoRecord();
            wmNoRecord.setTableId(item.getId());
            // 第三方交易单号
            wmNoRecord.setTradeNo(item.getOtherTradeNo());
            wmNoRecordList.add(wmNoRecord);
        }
        // 保存微盟交易单号记录表
        wmNoRecordService.saveBatch(wmNoRecordList);
        // 结束微盟交易明细导入
        redisTemplate.delete(SystemConstants.IMPORT_WM_TRADE_DETAIL_FLAG);
    }

    @Override
    public List<WmTradeDetailImportVo> findByIds(String idsStr) {
        if (!StringUtils.hasText(idsStr)) {
            return new ArrayList<>();
        }
        List<Long> ids = Arrays.stream(idsStr.split(",")).map(Long::parseLong).collect(Collectors.toList());
        List<WmTradeDetail> wmTradeDetailList = this.getBaseMapper().selectBatchIds(ids);
        List<WmTradeDetailImportVo> voList = new ArrayList<>();
        for (WmTradeDetail wmTradeDetail : wmTradeDetailList) {
            WmTradeDetailImportVo vo = new WmTradeDetailImportVo();
            BeanUtils.copyProperties(wmTradeDetail, vo);
            voList.add(vo);
        }
        return voList;
    }
}
